package edu.cornell.cs.lsi.mapreduce.pass4;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.URI;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class CCStats {
    private static enum CCStatsCounters {
        DISCRETE_COMPONENTS, TOTAL_COMPONENTS_SIZE
    }

    public static class SwapMapper extends Mapper<LongWritable, Text, Text, Text> {

        public void map(LongWritable key, Text value, Context context) throws InterruptedException, IOException {
            // swap vertex and parent
            String[] vertices = value.toString().trim().split("[\\s]+");
            double vertexX = Double.parseDouble(vertices[0]);
            double vertexY = Double.parseDouble(vertices[1]);
            double parentX = Double.parseDouble(vertices[2]);
            double parentY = Double.parseDouble(vertices[3]);
            context.write(new Text(parentX + " " + parentY), new Text(vertexX + " " + vertexY));
        }
    }

    public static class StatsReducer extends Reducer<Text, Text, Text, Text> {
        @SuppressWarnings("unused")
        public void reduce(Text parent, Iterable<Text> vertices, Context context) {
            context.getCounter(CCStatsCounters.DISCRETE_COMPONENTS).increment(1);
            int componentSize = 0;
            for (Text vertex : vertices) {
                componentSize++;
            }
            // divide by total number of vertices at the end to get expected component size
            context.getCounter(CCStatsCounters.TOTAL_COMPONENTS_SIZE).increment(componentSize * componentSize);
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Job job = new Job();
        job.setJobName("CCStats");
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        job.setJarByClass(CCStats.class);
        job.setMapperClass(SwapMapper.class);
        job.setReducerClass(StatsReducer.class);
        job.setNumReduceTasks(1);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(false);
        Counters counters = job.getCounters();
        Counter discreteComponents = counters.findCounter(CCStatsCounters.DISCRETE_COMPONENTS);
        Counter totalComponentSize = counters.findCounter(CCStatsCounters.TOTAL_COMPONENTS_SIZE);
        Path statsFile = new Path(args[2]);
        FileSystem fs = FileSystem.get(URI.create(args[2]), job.getConfiguration());
        BufferedReader statsFileIn = new BufferedReader(new InputStreamReader(fs.open(statsFile)));
        String verticesLine = statsFileIn.readLine();
        String edgesLine = statsFileIn.readLine();
        statsFileIn.close();
        // get average size by dividing by total num vertices, read vertices in and append stats
        PrintWriter statsFileOut = new PrintWriter(fs.create(statsFile, true));
        int totalVerticesNum = Integer.parseInt(verticesLine.split(":")[1].trim());
        // write previous stats back
        statsFileOut.println(verticesLine);
        statsFileOut.println(edgesLine);
        statsFileOut.println("Discrete Component Count " + discreteComponents.getValue());
        statsFileOut.println("Expected Component Size " + (double) totalComponentSize.getValue() / totalVerticesNum);
        statsFileOut.close();
    }
}
